package cn.doitedu.cn.doitedu.rtdw.etl;

import cn.doitedu.cn.doitedu.rtdw.udfs.MapToJson;
import cn.doitedu.cn.doitedu.rtdw.udfs.ToGeoHash;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

public class Job1_EventsLogCommonDimension {

    public static void main(String[] args) {
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 注册一个经纬度转geohash码的自定义函数
        tenv.createTemporaryFunction("to_geohash", ToGeoHash.class);
        tenv.createTemporaryFunction("map_json", MapToJson.class);



        // 1.建表，映射kafka中的 ods层的行为日志
        tenv.executeSql(
        " create table ods_events_kafka(         "+
                "    release_channel  string,       "+
                "    device_type string,            "+
                "    session_id string,             "+
                "    lat double,                    "+
                "    lng double,                    "+
                "    username string,               "+
                "    eventId string,                "+
                "    actionTime bigint,             "+
                "    properties map<string,string>, "+
                "    pt AS proctime()               "+  // 声明字段：表达式字段
                " ) with (                          "+
                "  'connector' = 'kafka',                             "+
                "  'topic' = 'ods-events-log',                        "+
                "  'properties.bootstrap.servers' = 'doitedu:9092',   "+
                "  'properties.group.id' = 'goo1',                    "+
                "  'scan.startup.mode' = 'latest-offset',             "+
                "  'value.format'='json',                             "+
                "  'value.json.fail-on-missing-field'='false',        "+
                "  'value.fields-include' = 'EXCEPT_KEY'              "+
                " )                                                   "
        );

        // 2. 建表，映射hbase中的用户注册信息表
        tenv.executeSql(
                " create table ums_member_hbase(           "+
                        "   username string,                       "+
                        " 	f row<id bigint,                       "+
                        " 	  phone string,                         "+
                        " 	  status int,                           "+
                        " 	  create_time timestamp(3),             "+
                        " 	  gender int,                           "+
                        " 	  birthday date,                        "+
                        "     city string,                         "+
                        "     job string,                          "+
                        "     source_type int>,                    "+
                        " 	PRIMARY KEY (username) NOT ENFORCED    "+
                        " ) with (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'dim_ums_member',        "+
                        "  'zookeeper.quorum' = '192.168.77.88:2181'     "+
                        " )                                        "
        );


        // 3. 建表，映射hbase中的geohash地理位置维表
        tenv.executeSql(
                " create table dim_geohash_hbase(              " +
                        "   geohash string,                       " +
                        " 	f row<                                " +
                        " 	  province string,                    " +
                        " 	  city string,                        " +
                        " 	  region string>,                     " +
                        " 	PRIMARY KEY (geohash) NOT ENFORCED    " +
                        " ) with (                                 " +
                        "  'connector' = 'hbase-2.2',              " +
                        "  'table-name' = 'dim_ref_geohash',        " +
                        "  'zookeeper.quorum' = '192.168.77.88:2181'     " +
                        " )                                        "
        );


        // 4.建表，映射hbase中的页面信息维表
        tenv.executeSql(
                " create table dim_pageinfo_hbase(              " +
                        "   url_prefix string,                     " +
                        " 	f row<                                " +
                        " 	  page_type string,                    " +
                        " 	  page_service string>,                " +
                        " 	PRIMARY KEY (url_prefix) NOT ENFORCED    " +
                        " ) with (                                 " +
                        "  'connector' = 'hbase-2.2',              " +
                        "  'table-name' = 'dim_pageurl_info',       " +
                        "  'zookeeper.quorum' = '192.168.77.88:2181'     " +
                        " )                                        "
        );


        // 5.建表，映射kafka中dwd层的行为明细
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                           "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "

                        +"     user_id           BIGINT,                     "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "

                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "

                        +"     page_url     STRING,                          "
                        +"     url_prefix    STRING,                          "
                        +"     page_type    STRING,                          "
                        +"     page_service STRING                           "
                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd-events-detail',                   "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'earliest-offset',         "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 6. 建表映射 doris 中接收明细数据的表
        tenv.executeSql(
                " CREATE TABLE user_events_detail_doris(            "+
                        "          dt                      DATE             "+
                        "         ,event_id                string           "+
                        "         ,user_id                 bigint           "+
                        "         ,gps_province            string           "+
                        "         ,gps_city                string           "+
                        "         ,gps_region              string           "+
                        "         ,username                string           "+
                        "         ,session_id              string           "+
                        "         ,event_time              bigint           "+
                        "         ,lat                     double           "+
                        "         ,lng                     double           "+
                        "         ,release_channel         string           "+
                        "         ,device_type             string           "+
                        "         ,properties              string           "+
                        "         ,register_phone          string           "+
                        "         ,user_status             int              "+
                        "         ,register_time           timestamp(3)     "+
                        "         ,register_gender         string           "+
                        "         ,register_birthday       DATE             "+
                        "         ,register_city           string           "+
                        "         ,register_job            string           "+
                        "         ,register_source_type    int              "+
                        "         ,page_url                string           "+
                        "         ,url_prefix              string           "+
                        "         ,page_type               string           "+
                        "         ,page_service            string           "+
                        " ) WITH (                                          "+
                        "    'connector' = 'doris',                         "+
                        "    'fenodes' = 'doitedu:8030',                    "+
                        "    'table.identifier' = 'dwd.user_events_detail', "+
                        "    'username' = 'root',                           "+
                        "    'password' = 'root',                           "+
                        "    'sink.label-prefix' = 'doris_label-xxx'        "+
                        " )                                                 "
        );

        // 7.写sql做维表关联
        tenv.executeSql(
                        "CREATE TEMPORARY VIEW joined AS  " +
                        "SELECT /*+ LOOKUP('table'='ums_member_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='5s','max-attempts'='3') */"+
                        " e.username                                                                                                                               "+
                        " ,e.session_id                                                                                                                            "+
                        " ,e.eventId as event_id                                                                                                                           "+
                        " ,e.actionTime as event_time                                                                                                                            "+
                        " ,e.lat                                                                                                                                   "+
                        " ,e.lng                                                                                                                                   "+
                        " ,e.release_channel                                                                             "+
                        " ,e.device_type                                                                                 "+
                        " ,e.properties           		                                                                 "+
                        " ,u.f.id as user_id                                                                                 "+
                        " ,u.f.phone as  register_phone                                                                      "+
                        " ,u.f.status as  user_status                                                                        "+
                        " ,u.f.create_time as register_time                                                                  "+
                        " ,u.f.gender as register_gender                                                                     "+
                        " ,u.f.birthday as register_birthday                                                                 "+
                        " ,u.f.city as register_city                                                                         "+
                        " ,u.f.job as register_job                                                                           "+
                      " ,u.f.source_type as register_source_type                                                             "+
                        " ,g.f.province as gps_province                                                                      "+
                        " ,g.f.city as gps_city                                                                              "+
                        " ,g.f.region as gps_region                                                                          "+
                        " ,e.properties['url']  as page_url                                                                  "+
                        " ,p.url_prefix                                                                                      "+
                        " ,p.f.page_type                                                                                     "+
                        " ,p.f.page_service                                                                                  "+
                        " from ods_events_kafka e                                                                            "+
                        " LEFT JOIN ums_member_hbase   FOR SYSTEM_TIME AS OF e.pt  AS u  on e.username = u.username                                         "+
                        " LEFT JOIN dim_geohash_hbase  FOR SYSTEM_TIME AS OF e.pt  AS g  on to_geohash(e.lat,e.lng) = g.geohash                             "+
                        " LEFT JOIN dim_pageinfo_hbase FOR SYSTEM_TIME AS OF e.pt  AS p  on regexp_extract(e.properties['url'],'(.*/).*?') = p.url_prefix     "
        );

        // 8将join结果插入kafka
        tenv.executeSql("insert into dwd_kafka select * from joined ");

        // 9. 写sql，将关联结果写入doris表
        tenv.executeSql(
                " INSERT INTO  user_events_detail_doris             "+
                        " SELECT                                            "+
                        "  to_date(date_format(to_timestamp_ltz(event_time,3),'yyyy-MM-dd'))  as  dt  "+
                        " ,event_id                                         "+
                        " ,user_id                                          "+
                        " ,gps_province                                     "+
                        " ,gps_city                                         "+
                        " ,gps_region                                       "+
                        " ,username                                         "+
                        " ,session_id                                       "+
                        " ,event_time                                       "+
                        " ,lat                                              "+
                        " ,lng                                              "+
                        " ,release_channel                                  "+
                        " ,device_type                                      "+
                        " ,map_json(properties) as  properties              "+
                        " ,register_phone                                   "+
                        " ,user_status                                      "+
                        " ,register_time                                    "+
                        " ,case register_gender  when 0 then '未知' when 1 then '男' when 2 then '女' end as register_gender  "+
                        " ,register_birthday                                "+
                        " ,register_city                                    "+
                        " ,register_job                                     "+
                        " ,register_source_type                             "+
                        " ,page_url                                         "+
                        " ,url_prefix                                       "+
                        " ,page_type                                        "+
                        " ,page_service                                     "+
                        " FROM joined                                       "
        );

    }
}
